
package com.huaweicloud.haydn.delivercore.agent.service.metadata.impl;

import static com.huaweicloud.haydn.delivercore.agent.config.Constants.DOT_FILETYPE_CSV;
import static com.huaweicloud.haydn.delivercore.agent.config.Constants.FILETYPE_PROPERTIES;
import static com.huaweicloud.haydn.delivercore.agent.config.Constants.FILETYPE_ZIP;
import static com.huaweicloud.haydn.delivercore.agent.config.Constants.SYS_URL;

import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.ExcelWriter;
import com.alibaba.excel.context.AnalysisContext;
import com.alibaba.excel.metadata.data.ReadCellData;
import com.alibaba.excel.read.listener.ReadListener;
import com.alibaba.excel.support.ExcelTypeEnum;
import com.alibaba.excel.write.metadata.WriteSheet;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.huaweicloud.haydn.delivercore.agent.common.AgentConstans;
import com.huaweicloud.haydn.delivercore.agent.common.DbTypeEnum;
import com.huaweicloud.haydn.delivercore.agent.common.SqlValidateUtils;
import com.huaweicloud.haydn.delivercore.agent.common.consistency.ConsistencySqlDefinition;
import com.huaweicloud.haydn.delivercore.agent.common.exception.BusinessErrorEnum;
import com.huaweicloud.haydn.delivercore.agent.common.exception.BusinessException;
import com.huaweicloud.haydn.delivercore.agent.config.DatabaseContext;
import com.huaweicloud.haydn.delivercore.agent.dto.CommonResult;
import com.huaweicloud.haydn.delivercore.agent.dto.excel.DbConsistencyExcelDto;
import com.huaweicloud.haydn.delivercore.agent.dto.excel.DbTableContentConsistencyExcelDto;
import com.huaweicloud.haydn.delivercore.agent.dto.haydn.ContentConsistencyConfigDto;
import com.huaweicloud.haydn.delivercore.agent.dto.haydn.DataMonitoringReportDto;
import com.huaweicloud.haydn.delivercore.agent.dto.haydn.DataParseDto;
import com.huaweicloud.haydn.delivercore.agent.dto.haydn.DbMonitoringObjResultDto;
import com.huaweicloud.haydn.delivercore.agent.dto.haydn.ExcelParseDto;
import com.huaweicloud.haydn.delivercore.agent.dto.haydn.MonitoringFieldMappingConfigDto;
import com.huaweicloud.haydn.delivercore.agent.entity.DbEntity;
import com.huaweicloud.haydn.delivercore.agent.entity.ZipFileEntity;
import com.huaweicloud.haydn.delivercore.agent.service.CommonUtils;
import com.huaweicloud.haydn.delivercore.agent.service.HttpRequestService;
import com.huaweicloud.haydn.delivercore.agent.service.metadata.ConsistencyMonitoringService;
import com.huaweicloud.haydn.delivercore.agent.service.metadata.DatabaseUtil;
import com.huaweicloud.haydn.delivercore.agent.utils.DateUtil;
import com.huaweicloud.haydn.delivercore.agent.utils.HiveAuthUtil;
import com.huaweicloud.haydn.delivercore.agent.utils.SSLUtils;
import com.huaweicloud.haydn.delivercore.agent.utils.ZipUtil;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.stream.Collectors;
import java.util.zip.ZipEntry;
import java.util.zip.ZipFile;

import javax.annotation.PostConstruct;

/**
 * 功能描述
 *
 * @since 2023-02-01
 */
@Service
@Slf4j
public class ConsistencyMonitoringServiceImpl implements ConsistencyMonitoringService {
    @Autowired
    private DatabaseContext databaseContext;

    @Autowired
    private HttpRequestService httpRequestService;

    @Autowired
    private ImplementMonitoringServiceImpl implementMonitoringService;

    @Autowired
    private CommonUtils commonUtils;

    @Autowired
    private ZipUtil zipUtil;

    @PostConstruct
    public void postConstruct() {
        SSLUtils.disableSslVerification();
    }

    @Override
    public void genTableFieldData() {
        genDataType(AgentConstans.ConsistencyTypeEnum.TABLE_FIELDS);
    }

    @Override
    public void genTableRowsData() {
        genDataType(AgentConstans.ConsistencyTypeEnum.TABLE_ROWS);
    }

    @Override
    public void genTableContentData() {
        genDataType(AgentConstans.ConsistencyTypeEnum.TABLE_CONTENTS);
    }

    public void genDataType(AgentConstans.ConsistencyTypeEnum type) {
        try (InputStream inputStream =
            new FileInputStream(System.getProperty(SYS_URL) + File.separator + "application." + FILETYPE_PROPERTIES)) {
            Properties monitorConfig = new Properties();
            monitorConfig.load(inputStream);
            List<ZipFileEntity> list = new ArrayList<>();
            for (int i = 0; i < 100; i++) {
                if (!read(monitorConfig, (i + 1))) {
                    log.info("break at i = " + i + ", idx = " + (i + 1));
                    break;
                }
                ZipFileEntity zipFile = genDataInPool(i + 1, monitorConfig, type);
                if (Objects.nonNull(zipFile)) {
                    list.add(zipFile);
                }
            }
            zipUtil.execZip(list, "monitorConsistencyResult");
        } catch (IOException e) {
            log.error(e.getMessage());
        }

    }

    private boolean read(Properties dataSourceConfig, int i) {
        String dbType = dataSourceConfig.getProperty("monitor.target.consistency.dbtype." + i);
        String hosts = dataSourceConfig.getProperty("monitor.target.consistency.hosts." + i);
        String port = dataSourceConfig.getProperty("monitor.target.consistency.port." + i);
        String userName = dataSourceConfig.getProperty("monitor.target.consistency.username." + i);
        String password = dataSourceConfig.getProperty("monitor.target.consistency.password." + i);
        if (dbType == null) {
            return false;
        }
        boolean proBase;
        if (DbTypeEnum.getValueByName(dbType).equals(DbTypeEnum.MRS_HUDI)) {
            proBase = StringUtils.isNotEmpty(dbType) && StringUtils.isNotEmpty(hosts);
        } else {
            proBase = StringUtils.isNotEmpty(dbType) && StringUtils.isNotEmpty(hosts) && StringUtils.isNotEmpty(port);
        }

        boolean proSuper = StringUtils.isNotEmpty(userName) && StringUtils.isNotEmpty(password);
        return proBase && proSuper;
    }

    private DbEntity getSourceDb(int idx, Properties dataSourceConfig, AgentConstans.ConsistencyTypeEnum type) {
        String dbType = dataSourceConfig.getProperty("monitor.source.consistency.dbtype." + idx);
        String hosts = dataSourceConfig.getProperty("monitor.source.consistency.hosts." + idx);
        String port = dataSourceConfig.getProperty("monitor.source.consistency.port." + idx);
        String userName = dataSourceConfig.getProperty("monitor.source.consistency.username." + idx);
        String password = dataSourceConfig.getProperty("monitor.source.consistency.password." + idx);
        String databaseName = dataSourceConfig.getProperty("monitor.source.consistency.database." + idx);
        String isSafeMode = dataSourceConfig.getProperty("monitor.source.consistency.safemode." + idx);
        DbEntity sourceDb = getDbEntity(dbType, hosts, port, userName, password, databaseName, isSafeMode);
        DatabaseUtil sourceDbUtil =
            databaseContext.getStrategy(DbTypeEnum.getDatabaseUtilComponentByConfigType(dbType));
        sourceDbUtil.init(hosts, port, databaseName, userName, password, dbType, "", "");
        sourceDb.setDatabaseUtil(sourceDbUtil);
        String sourceSchemaSelected =
            Optional.ofNullable(dataSourceConfig.getProperty("monitor.source.consistency.schema." + idx)).orElse("");
        String sourceTableSelected =
            Optional.ofNullable(dataSourceConfig.getProperty("monitor.source.consistency.table." + idx)).orElse("");
        String timestamp =
            Optional.ofNullable(dataSourceConfig.getProperty("monitor.target.consistency.rownum.timestamp." + idx))
                .orElse("");
        String ignoreField =
            Optional.ofNullable(dataSourceConfig.getProperty("monitor.target.consistency.ignore.field." + idx))
                .orElse("");
        sourceDb.setSchemaSelected(sourceSchemaSelected);
        sourceDb.setTableSelected(sourceTableSelected);
        sourceDb.setTimeStamp(timestamp);
        sourceDb.setIgnoreField(ignoreField);
        return sourceDb;
    }

    private DbEntity getTargetDb(int idx, Properties dataSourceConfig, AgentConstans.ConsistencyTypeEnum type) {
        String dbType = dataSourceConfig.getProperty("monitor.target.consistency.dbtype." + idx);
        String hosts = dataSourceConfig.getProperty("monitor.target.consistency.hosts." + idx);
        String port = dataSourceConfig.getProperty("monitor.target.consistency.port." + idx);
        String userName = dataSourceConfig.getProperty("monitor.target.consistency.username." + idx);
        String password = dataSourceConfig.getProperty("monitor.target.consistency.password." + idx);
        String databaseName = dataSourceConfig.getProperty("monitor.target.consistency.database." + idx);
        String isSafeMode = dataSourceConfig.getProperty("monitor.target.consistency.safemode." + idx);
        DbEntity targetDb = getDbEntity(dbType, hosts, port, userName, password, databaseName, isSafeMode);
        DatabaseUtil targetDbUtil =
            databaseContext.getStrategy(DbTypeEnum.getDatabaseUtilComponentByConfigType(dbType));
        if ("MRS-Hudi".equals(dbType) && "true".equals(isSafeMode)) {
            databaseName = HiveAuthUtil.getHiveAuthJDBC(userName, databaseName);
        }
        targetDbUtil.init(hosts, port, databaseName, userName, password, dbType, "", "");
        targetDb.setDatabaseUtil(targetDbUtil);
        String targetSchemaSelected =
            Optional.ofNullable(dataSourceConfig.getProperty("monitor.target.consistency.schema." + idx)).orElse("");
        String targetTableSelected =
            Optional.ofNullable(dataSourceConfig.getProperty("monitor.target.consistency.table." + idx)).orElse("");
        String timestamp =
            Optional.ofNullable(dataSourceConfig.getProperty("monitor.target.consistency.rownum.timestamp." + idx))
                .orElse("");
        String ignoreField =
            Optional.ofNullable(dataSourceConfig.getProperty("monitor.target.consistency.ignore.field." + idx))
                .orElse("");
        targetDb.setSchemaSelected(targetSchemaSelected);
        targetDb.setTableSelected(targetTableSelected);
        targetDb.setTimeStamp(timestamp);
        targetDb.setIgnoreField(ignoreField);
        String uuid = "";
        if (AgentConstans.ConsistencyTypeEnum.TABLE_ROWS.equals(type)) {
            uuid = dataSourceConfig.getProperty("monitor.target.datamodel.consistencytablerows.uuid." + idx);
        } else if (AgentConstans.ConsistencyTypeEnum.TABLE_FIELDS.equals(type)) {
            uuid = dataSourceConfig.getProperty("monitor.target.datamodel.consistencytablefield.uuid." + idx);
        } else {
            uuid = dataSourceConfig.getProperty("monitor.target.datamodel.consistencytablecontents.uuid." + idx);
        }
        targetDb.setUuid(uuid);
        return targetDb;
    }

    private DbEntity getDbEntity(String dbType, String hosts, String port, String userName, String password,
        String databaseName, String isSafeMode) {
        DbEntity db = new DbEntity();
        db.setDbType(dbType);
        db.setHosts(hosts);
        db.setPort(port);
        db.setUserName(userName);
        db.setPassword(password);
        db.setDatabaseName(databaseName);
        db.setIsSafeMode(isSafeMode);
        return db;
    }

    private ZipFileEntity genDataInPool(int idx, Properties dataSourceConfig, AgentConstans.ConsistencyTypeEnum type) {
        // 目的端
        DbEntity targetDb = getTargetDb(idx, dataSourceConfig, type);
        // 源端
        DbEntity sourceDb = getSourceDb(idx, dataSourceConfig, type);

        log.info("Successfully read the configuration file and successfully initialized!");
        ZipFileEntity zipFile = null;
        try (Connection targetConn = targetDb.getDatabaseUtil().getConnection();
            Connection sourceConn = sourceDb.getDatabaseUtil().getConnection()) {
            if (AgentConstans.ConsistencyTypeEnum.TABLE_ROWS.equals(type)
                || AgentConstans.ConsistencyTypeEnum.TABLE_FIELDS.equals(type)) {
                if (DbTypeEnum.DWS.name().equals(targetDb.getDbType())
                    || DbTypeEnum.MRS_HUDI.name().equals(targetDb.getDbType())) {
                    zipFile = genDbConsistencyExcelDto(targetDb, sourceDb, targetConn, sourceConn, idx + "", type);
                }
            }

            if (AgentConstans.ConsistencyTypeEnum.TABLE_CONTENTS.equals(type)) {
                // 获取表内容一致性对比所需配置参数
                // todo 待优化配置问题读取，同个数据库连接只创建一次
                ContentConsistencyConfigDto dbContentConsistencyConfig = new ContentConsistencyConfigDto();
                boolean isContent = getDbContentConsistencyConfig(idx, dataSourceConfig, targetDb, sourceDb,
                    dbContentConsistencyConfig);
                if (!isContent) {
                    return null;
                }
                zipFile = genTableContentConsistencyExcels(targetDb, sourceDb, targetConn, sourceConn, idx + "",
                    dbContentConsistencyConfig);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new BusinessException(BusinessErrorEnum.BASE_INTERNAL_SERVER_ERROR, "Failed to generate excel!");
        }
        return zipFile;
    }

    private String getDbTableRowsSqlByDbType(String dbType, String dbNm, String schemaNm, String tableNm,
        String timestampNm) {
        String sql = "";
        if ("dws".equals(dbType)) {
            sql = String.format("select count(*) from %s.%s", schemaNm, tableNm);
            if (StringUtils.isNotEmpty(timestampNm)) {
                sql = sql + " where " + "\"" + timestampNm + "\"" + " <= ?";
            }
        } else if ("oracle".equals(dbType)) {
            sql = String.format("select count(*) from %s.%s", schemaNm, "\"" + tableNm + "\"");
            if (StringUtils.isNotEmpty(timestampNm)) {
                sql = sql + " where " + timestampNm.toUpperCase(Locale.ROOT) + " <= ?";
            }
        } else if ("mysql".equals(dbType)) {
            sql = String.format("select count(*) from %s.%s", dbNm, tableNm);
            if (StringUtils.isNotEmpty(timestampNm)) {
                sql = sql + " where " + timestampNm + " <= ?";
            }
        } else if ("MRS-Hudi".equals(dbType)) {
            sql = String.format("select count(1) from %s", tableNm);
            if (StringUtils.isNotEmpty(timestampNm)) {
                sql = sql + " where " + timestampNm + " <= ?";
            }
        }

        return sql;
    }

    /**
     * 功能描述:获取表的内容一致性所特有的字段
     *
     * @since 2023-03-17
     */
    private boolean getDbContentConsistencyConfig(int idx, Properties dataSourceConfig, DbEntity targetDb,
        DbEntity sourceDb, ContentConsistencyConfigDto configDto) {
        // 获取表内容一致性检测相关的信息
        String contenMappingsJson = dataSourceConfig.getProperty("monitor.target.consistency.field.mapping." + idx);
        String filterSqlStatement = dataSourceConfig.getProperty("monitor.target.consistency.filer.sql." + idx);
        String singleExtractSize = dataSourceConfig.getProperty("monitor.target.consistency.extract.size." + idx);
        String extractBatch = dataSourceConfig.getProperty("monitor.target.consistency.extract.batch." + idx);
        // 校验读取的过滤条件（校验是否含有delete、update等敏感字段）
        SqlValidateUtils.validSqlBadWords(filterSqlStatement);

        List<MonitoringFieldMappingConfigDto> mappingConfigDtos =
            JSONObject.parseArray(contenMappingsJson, MonitoringFieldMappingConfigDto.class);
        // 字段内容映射不允许为空,为空的为非表内容一致性检测
        if (CollectionUtils.isEmpty(mappingConfigDtos)) {
            return false;
        }

        DbTypeEnum sourceDbType = DbTypeEnum.getValueByName(sourceDb.getDbType());
        DbTypeEnum targetDbType = DbTypeEnum.getValueByName(targetDb.getDbType());

        configDto.setMonitoringFieldMappings(mappingConfigDtos);
        configDto.setFilterSqlStatement(filterSqlStatement);
        configDto.setExtractBatch(Integer.parseInt(extractBatch));
        configDto.setSingleExtractSize(Integer.parseInt(singleExtractSize));
        configDto.setSourceDbType(sourceDbType);
        configDto.setTargetDbType(targetDbType);

        return true;
    }

    /**
     * 功能描述: 获取表内容一致性对比结果
     *
     * @since 2023-03-22
     */
    private ZipFileEntity genTableContentConsistencyExcels(DbEntity targetDb, DbEntity sourceDb, Connection targetConn,
        Connection sourceConn, String idx, ContentConsistencyConfigDto configDto) {
        String fileName = idx + "_" + AgentConstans.ConsistencyTypeEnum.TABLE_CONTENTS.getConsistencyType() + "_"
            + new SimpleDateFormat("yyyyMMdd").format(new Date()) + DOT_FILETYPE_CSV;

        List<DbTableContentConsistencyExcelDto> allContentResults = new ArrayList<>();
        // todo 待用多线程优化
        for (int round = 1; round <= configDto.getExtractBatch(); round++) {
            // 获取源端数据表中所有的值
            String sourceContentSql = ConsistencySqlDefinition.getSourceContentSql(sourceDb.getTableSelected(), round,
                configDto, sourceDb.getDatabaseName(), sourceDb.getSchemaSelected());
            List<List<String>> sourceTableValues = getSourceTableValues(sourceConn, sourceContentSql, configDto);
            // 根据源端数据对比目的端数据并生成结果记录
            List<String> notFoundSourceDataValues =
                getTargetDataAndValid(sourceDb, targetConn, targetDb, configDto, sourceTableValues);
            allContentResults.add(
                genTableContentResult(targetDb, sourceDb, round, sourceTableValues.size(), notFoundSourceDataValues));
            log.info(
                "Source table:{} of source database:{}, Table table:{} of target database:{}, actual total siz:{}, not found data total size:{}",
                sourceDb.getTableSelected(), sourceDb.getDatabaseName(), targetDb.getTableSelected(),
                targetDb.getDatabaseName(), sourceTableValues.size(), notFoundSourceDataValues.size());
        }

        ExcelWriter excelWriter = EasyExcel.write(fileName, DbTableContentConsistencyExcelDto.class).build();
        WriteSheet writeSheet = EasyExcel.writerSheet("sheet").build();
        excelWriter.write(allContentResults, writeSheet);
        excelWriter.finish();
        ZipFileEntity zipFile = new ZipFileEntity();
        zipFile.setFileName(fileName);
        zipFile.setType(AgentConstans.ConsistencyTypeEnum.TABLE_CONTENTS.getConsistencyType());
        return zipFile;
    }

    /**
     * 功能描述: 根据 目的端和源端不一致的数据生成表内容对比的EXCEL结果
     *
     * @since 2023-03-23
     */
    private DbTableContentConsistencyExcelDto genTableContentResult(DbEntity targetDb, DbEntity sourceDb, int round,
        int atualSize, List<String> notFoundSourceDataValues) {

        DbTableContentConsistencyExcelDto excelDto = new DbTableContentConsistencyExcelDto();
        excelDto.setMonitoringObjectUuid(targetDb.getUuid());
        excelDto.setTargetDbName(targetDb.getDatabaseName());
        excelDto.setTargetSchemaName(targetDb.getSchemaSelected());
        excelDto.setTargetTableName(targetDb.getTableSelected());
        excelDto.setSourceDbName(sourceDb.getDatabaseName());
        excelDto.setSourceSchemaName(sourceDb.getSchemaSelected());
        excelDto.setSourceTableName(sourceDb.getTableSelected());
        excelDto.setBatchActualSize(atualSize);
        excelDto.setNotFoundDataSize(notFoundSourceDataValues.size());
        excelDto.setRound(round);
        return excelDto;

    }

    /**
     * 功能描述: 根据源端数据对比目的端数据并生成结果记录
     *
     * @since 2023-03-23
     */
    private List<String> getTargetDataAndValid(DbEntity sourceDb, Connection targetConn, DbEntity targetDb,
        ContentConsistencyConfigDto configDto, List<List<String>> sourceTableValues) {
        // 生成一致性比对的sql(预编译过)
        String targetContentSql = ConsistencySqlDefinition.getTargetContentSql(targetDb.getTableSelected(), configDto,
            targetDb.getDatabaseName(), targetDb.getSchemaSelected());

        List<String> sourceFiledNames = configDto.getMonitoringFieldMappings()
            .stream()
            .map(MonitoringFieldMappingConfigDto::getSourceFieldNameEn)
            .collect(Collectors.toList());

        // 未查找到的源端数据列表
        List<String> notFoundSouceDataList = new ArrayList<>();

        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            ps = targetConn.prepareStatement(targetContentSql, ResultSet.TYPE_SCROLL_INSENSITIVE,
                ResultSet.CONCUR_READ_ONLY);
            for (List<String> perLineSourceData : sourceTableValues) {
                for (int columnIndex = 1; columnIndex <= configDto.getMonitoringFieldMappings().size(); columnIndex++) {
                    if (perLineSourceData.get(columnIndex - 1) != null) {
                        ps.setString(columnIndex, perLineSourceData.get(columnIndex - 1));
                    }
                }
                rs = ps.executeQuery();
                // 未找到对应的源端数据，则将其记录结果
                if (!rs.next()) {
                    StringBuilder sourceLineDate = new StringBuilder();
                    sourceLineDate.append("{");
                    for (int columnIndex = 1; columnIndex <= configDto.getMonitoringFieldMappings().size();
                        columnIndex++) {
                        sourceLineDate.append(String.format("%s:%s", sourceFiledNames.get(columnIndex - 1),
                            perLineSourceData.get(columnIndex - 1)));
                        String separator = columnIndex == configDto.getMonitoringFieldMappings().size() ? "}" : ", ";
                        sourceLineDate.append(separator);
                    }
                    notFoundSouceDataList.add(sourceLineDate.toString());
                    log.error("Source data:{} in database:{}, table :{} not found!", sourceDb.getDatabaseName(),
                        sourceDb.getTableSelected(), sourceLineDate);
                }
            }

        } catch (SQLException throwables) {
            log.error(
                "An exception occurred when executing the target SQL statement:{}. Please check the SQL statement.",
                targetContentSql);
        } finally {
            try {
                if (rs != null) {
                    rs.close();
                }
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException e) {
                log.error("close db failed: ", e);
            }

        }
        return notFoundSouceDataList;
    }

    /**
     * 功能描述: 获取源端数据表中所有的值
     *
     * @since 2023-03-23
     */
    private List<List<String>> getSourceTableValues(Connection sourceConn, String sourceContentSql,
        ContentConsistencyConfigDto configDto) {
        // 存放源端数据结果集的list
        List<List<String>> sourceValueList = new ArrayList<>();

        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            ps = sourceConn.prepareStatement(sourceContentSql, ResultSet.TYPE_SCROLL_INSENSITIVE,
                ResultSet.CONCUR_READ_ONLY);
            rs = ps.executeQuery();
            while (rs.next()) {
                List<String> singleLineValue = new ArrayList<>();
                for (int columnIndex = 1; columnIndex <= configDto.getMonitoringFieldMappings().size(); columnIndex++) {
                    singleLineValue.add(rs.getString(columnIndex));
                }
                sourceValueList.add(singleLineValue);
            }
        } catch (SQLException throwables) {
            log.error(
                "An exception occurred when executing the source SQL statement:{}. Please check the SQL statement.",
                sourceContentSql);
        } finally {
            try {
                if (rs != null) {
                    rs.close();
                }
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException e) {
                log.error("close db failed: ", e);
            }

        }
        return sourceValueList;
    }

    /**
     * 功能描述: 获取表内容一致性对比结果
     *
     * @since 2023-03-22
     */
    private ZipFileEntity genDbConsistencyExcelDto(DbEntity targetDb, DbEntity sourceDb, Connection targetConn,
        Connection sourceConn, String idx, AgentConstans.ConsistencyTypeEnum type) {
        if (StringUtils.isBlank(targetDb.getTableSelected())) {
            return null;
        }
        List<String> tables = new ArrayList<>();
        CollectionUtils.addAll(tables, targetDb.getTableSelected().split(","));
        if ("dws".equals(targetDb.getDbType())) {
            for (int i = 0; i < tables.size(); i++) {
                tables.set(i, targetDb.getSchemaSelected() + "." + tables.get(i));
            }
        }
        List<DbConsistencyExcelDto> dbConsistencyExcelDtos =
            getDbData(targetDb, sourceDb, targetConn, sourceConn, tables);
        HashSet<DbConsistencyExcelDto> rowExcelDtos = new HashSet<>(dbConsistencyExcelDtos);
        String fileName = idx + "_" + type.getConsistencyType() + "_"
            + new SimpleDateFormat("yyyyMMdd").format(new Date()) + DOT_FILETYPE_CSV;
        ExcelWriter excelWriter = EasyExcel.write(fileName, DbConsistencyExcelDto.class).build();
        WriteSheet writeSheet = EasyExcel.writerSheet("sheet").build();
        excelWriter.write(rowExcelDtos, writeSheet);
        excelWriter.finish();
        ZipFileEntity zipFile = new ZipFileEntity();
        zipFile.setFileName(fileName);
        zipFile.setType(type.getConsistencyType());
        return zipFile;
    }

    private List<DbConsistencyExcelDto> getDbData(DbEntity targetDb, DbEntity sourceDb, Connection targetConn,
        Connection sourceConn, List<String> tables) {
        List<DbConsistencyExcelDto> dbConsistencyExcelDtos = new ArrayList<>();
        for (int i = 0; i < tables.size(); i++) {
            String schema;
            String tableNm;
            if ("dws".equals(targetDb.getDbType())) {
                String[] tableArr = tables.get(i).split("\\.");
                schema = tableArr[0];
                tableNm = tableArr[1];
            } else {
                schema = null;
                tableNm = tables.get(i);
            }
            DbConsistencyExcelDto dbConsistencyExcelDto = new DbConsistencyExcelDto();
            dbConsistencyExcelDto.setMonitoringObjectUuid(targetDb.getUuid());
            dbConsistencyExcelDto.setTargetDbName(targetDb.getDatabaseName());
            dbConsistencyExcelDto.setTargetSchemaName(schema);
            dbConsistencyExcelDto.setTargetTableName(tableNm);
            getDbRows(targetDb, targetConn, schema, tableNm, dbConsistencyExcelDto);
            getDbFields(targetDb, targetConn, schema, tableNm, dbConsistencyExcelDto);
            if ("oracle".equals(sourceDb.getDbType())) {
                genDataOracleRowsAndFields(sourceDb, sourceConn, i, dbConsistencyExcelDto);
            } else if ("mysql".equals(sourceDb.getDbType())) {
                genDataMysqlRowsAndFields(sourceDb, sourceConn, i, dbConsistencyExcelDto);
            }
            dbConsistencyExcelDtos.add(dbConsistencyExcelDto);
        }
        return dbConsistencyExcelDtos;
    }

    private void getDbFields(DbEntity dbEntity, Connection conn, String schema, String tableNm,
        DbConsistencyExcelDto dbConsistencyExcelDto) {
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            List<String> fileds = new ArrayList<>();
            if ("MRS-Hudi".equals(dbEntity.getDbType())) {
                String sql = String.format("desc %s.%s", dbEntity.getDatabaseName(), tableNm);
                ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
                rs = ps.executeQuery();
                while (rs.next()) {
                    fileds.add(rs.getString(1));
                }
                fileds.removeAll(Arrays.asList("_hoodie_commit_seqno", "_hoodie_file_name", "_hoodie_partition_path",
                    "_hoodie_record_key", "_hoodie_commit_time"));
            } else {
                String sql = dbEntity.getDatabaseUtil().getAllFieldsSql();
                ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
                ps.setString(1, schema + "." + tableNm);
                rs = ps.executeQuery();
                while (rs.next()) {
                    implementMonitoringService.dwsField(fileds, rs.getString(1));
                }
            }
            List<String> filedsList = new ArrayList<>();
            for (int i = 0; i < fileds.size(); i++) {
                String field = fileds.get(i);
                if (StringUtils.isNotEmpty(dbEntity.getIgnoreField()) && dbEntity.getIgnoreField().contains(field)) {
                    continue;
                }
                filedsList.add(field);
            }
            dbConsistencyExcelDto.setTargetFieldName(String.join("&", filedsList));
        } catch (SQLException e) {
            log.error(e.getMessage(), e);
            throw new BusinessException(BusinessErrorEnum.BASE_INTERNAL_SERVER_ERROR);
        } finally {
            try {
                if (rs != null) {
                    rs.close();
                }
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException e) {
                log.error("close db failed: ", e);
            }
        }
    }

    private void getDbRows(DbEntity dbEntity, Connection conn, String schema, String tableNm,
        DbConsistencyExcelDto dbConsistencyExcelDto) {
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            String sql = getDbTableRowsSqlByDbType(dbEntity.getDbType(), dbEntity.getDatabaseName(), schema, tableNm,
                dbEntity.getTimeStamp());
            ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            if (StringUtils.isNotEmpty(dbEntity.getTimeStamp())) {
                ps.setString(1, DateUtil.getPreDay());
            }
            rs = ps.executeQuery();
            while (rs.next()) {
                dbConsistencyExcelDto.setTargetTableRow(rs.getString(1));
            }
        } catch (SQLException e) {
            log.error(e.getMessage(), e);
            throw new BusinessException(BusinessErrorEnum.BASE_INTERNAL_SERVER_ERROR);
        } finally {
            try {
                if (rs != null) {
                    rs.close();
                }
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException e) {
                log.error("close db failed: ", e);
            }
        }
    }

    private void genDataMysqlRowsAndFields(DbEntity sourceDb, Connection conn, int n,
        DbConsistencyExcelDto dbConsistencyExcelDto) {
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            List<String> tables = new ArrayList<>();
            if (StringUtils.isBlank(sourceDb.getTableSelected())) {
                return;
            }
            CollectionUtils.addAll(tables, sourceDb.getTableSelected().split(","));
            String table = tables.get(n);
            if (StringUtils.isNotEmpty(table)) {
                String sql1 = getDbTableRowsSqlByDbType(sourceDb.getDbType(), sourceDb.getDatabaseName(), null, table,
                    sourceDb.getTimeStamp());
                ps = conn.prepareStatement(sql1, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
                if (StringUtils.isNotEmpty(sourceDb.getTimeStamp())) {
                    ps.setString(1, DateUtil.getPreDay());
                }
                rs = ps.executeQuery();
                while (rs.next()) {
                    dbConsistencyExcelDto.setSourceTableRow(rs.getString(1));
                }
                dbConsistencyExcelDto.setSourceTableName(table);

                String sql2 = sourceDb.getDatabaseUtil()
                    .getAllFieldsSql(commonUtils.sqlQueryBuild(sourceDb.getDatabaseName()),
                        commonUtils.sqlQueryBuild(table));
                ps = conn.prepareStatement(sql2, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
                rs = ps.executeQuery();
                List<String> fileds = new ArrayList<>();
                while (rs.next()) {
                    String field = rs.getString(3);
                    if (StringUtils.isNotEmpty(sourceDb.getIgnoreField())
                        && sourceDb.getIgnoreField().contains(field)) {
                        continue;
                    }
                    fileds.add(field);
                }
                dbConsistencyExcelDto.setSourceFieldName(String.join("&", fileds));
            }
        } catch (SQLException e) {
            log.error(e.getMessage(), e);
            throw new BusinessException(BusinessErrorEnum.BASE_INTERNAL_SERVER_ERROR);
        } finally {
            try {
                if (rs != null) {
                    rs.close();
                }
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException e) {
                log.error("close db failed: ", e);
            }
        }
    }

    private void genDataOracleRowsAndFields(DbEntity sourceDb, Connection sourceConn, int n,
        DbConsistencyExcelDto dbConsistencyExcelDto) {
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            dbConsistencyExcelDto.setSourceDbName(sourceDb.getDatabaseName());
            if (StringUtils.isBlank(sourceDb.getTableSelected())) {
                return;
            }
            List<String> oracleTables = new ArrayList<>();
            CollectionUtils.addAll(oracleTables, sourceDb.getTableSelected().split(","));
            for (int i = 0; i < oracleTables.size(); i++) {
                oracleTables.set(i, sourceDb.getSchemaSelected() + "." + oracleTables.get(i));
            }
            String schemaAndTableNm = oracleTables.get(n);
            if (StringUtils.isNotEmpty(schemaAndTableNm)) {
                String[] arr = schemaAndTableNm.split("\\.");
                String oracleSchema = arr[0];
                String oracleTableNm = arr[1];
                try {
                    String sql = getDbTableRowsSqlByDbType(sourceDb.getDbType(), sourceDb.getDatabaseName(),
                        oracleSchema, oracleTableNm, sourceDb.getTimeStamp());
                    ps = sourceConn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE,
                        ResultSet.CONCUR_READ_ONLY);
                    if (StringUtils.isNotEmpty(sourceDb.getTimeStamp())) {
                        ps.setString(1, DateUtil.getPreDay());
                    }
                    rs = ps.executeQuery();
                    while (rs.next()) {
                        dbConsistencyExcelDto.setSourceTableRow(rs.getString(1));
                    }
                } catch (SQLException e) {
                    throw new BusinessException(BusinessErrorEnum.BASE_INTERNAL_SERVER_ERROR);
                }
                dbConsistencyExcelDto.setSourceSchemaName(oracleSchema);
                dbConsistencyExcelDto.setSourceTableName(oracleTableNm);

                String filedsSql =
                    String.format("SELECT column_name FROM dba_tab_columns where table_name ='%s'", oracleTableNm);
                ps = sourceConn.prepareStatement(filedsSql, ResultSet.TYPE_SCROLL_INSENSITIVE,
                    ResultSet.CONCUR_READ_ONLY);
                rs = ps.executeQuery();
                List<String> fileds = new ArrayList<>();
                while (rs.next()) {
                    String field = rs.getString(1);
                    if (StringUtils.isNotEmpty(sourceDb.getIgnoreField())
                        && sourceDb.getIgnoreField().contains(field)) {
                        continue;
                    }
                    fileds.add(field);
                }
                dbConsistencyExcelDto.setSourceFieldName(String.join("&", fileds));
            }
        } catch (SQLException e) {
            log.error(e.getMessage(), e);
            throw new BusinessException(BusinessErrorEnum.BASE_INTERNAL_SERVER_ERROR);
        } finally {
            try {
                if (rs != null) {
                    rs.close();
                }
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException e) {
                log.error("close db failed: ", e);
            }
        }
    }

    private String getSchemaAndTableNm(List<String> tables, String tableNm) {
        String res = "";
        for (int i = 0; i < tables.size(); i++) {
            if (tables.get(i).contains(tableNm)) {
                res = tables.get(i);
                break;
            }
        }
        return res;
    }

    public List<String> getOracleAllTables(DbEntity sourceDb, Connection conn) {
        List<String> tables = new ArrayList<>();
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            String sql =
                "select * from dba_tables where owner in" + commonUtils.sqlQueryBuild(sourceDb.getSchemaSelected());
            ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            rs = ps.executeQuery();
            while (rs.next()) {
                tables.add(rs.getString(1) + "." + rs.getString(2));
            }
            return tables;
        } catch (SQLException e) {
            throw new BusinessException(BusinessErrorEnum.BASE_INTERNAL_SERVER_ERROR);
        } finally {
            try {
                if (rs != null) {
                    rs.close();
                }
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException e) {
                log.error("close db failed: ", e);
            }
        }
    }

    @Override
    public void consistencyParseAndUpload(DataParseDto parseDto) {
        ZipFile zip = null;
        try {
            zip = new ZipFile(new File(System.getProperty(SYS_URL), "monitorConsistencyResult." + FILETYPE_ZIP),
                StandardCharsets.UTF_8);
            Enumeration<? extends ZipEntry> entries = zip.entries();
            List<DataMonitoringReportDto> dataMonitoringReportDtos = new ArrayList<>();
            while (entries.hasMoreElements()) {
                parseZip(zip, entries, dataMonitoringReportDtos);
            }
            log.info("Successfully parsed the compressed package!");
            HttpHeaders headers = new HttpHeaders();
            headers.set("Content-Type", "application/json;charset=utf8");
            headers.set("access-code", parseDto.getAccessCode());
            headers.set("deliver-project-id", parseDto.getProjectId());
            for (int i = 0; i < dataMonitoringReportDtos.size(); i++) {
                HttpEntity<String> entity =
                    new HttpEntity<>(new ObjectMapper().writeValueAsString(dataMonitoringReportDtos.get(i)), headers);
                ResponseEntity<CommonResult> resp =
                    httpRequestService.execute(parseDto.getUrl(), HttpMethod.POST, entity, CommonResult.class);
                if (resp.getBody() == null) {
                    throw new BusinessException(BusinessErrorEnum.BASE_INTERNAL_SERVER_ERROR, "response is null!");
                }
            }
        } catch (IOException | ParseException e) {
            log.error(e.getMessage());
        } finally {
            if (zip != null) {
                try {
                    zip.close();
                } catch (IOException e) {
                    log.error("Close zip failed!");
                    log.error(e.getMessage(), e);
                }
            }
        }
    }

    private void parseZip(ZipFile zip, Enumeration<? extends ZipEntry> entries,
        List<DataMonitoringReportDto> dataMonitoringReportDtos) throws ParseException, IOException {
        ZipEntry zipEntry = entries.nextElement();
        String csvFullName = zipEntry.getName();
        String csvName = csvFullName.substring(0, csvFullName.lastIndexOf(".")).toLowerCase(Locale.US);
        implementMonitoringService.csvNameCheck(csvName);
        String[] split = csvName.split("_");
        String idx = split[0];
        String typeName = split[1];
        String genDate = split[2];
        if (!"csv".equals(csvFullName.substring(csvFullName.lastIndexOf(".") + 1).toLowerCase(Locale.US))) {
            throw new BusinessException(BusinessErrorEnum.BASE_INTERNAL_SERVER_ERROR);
        }
        ExcelParseDto excelParseDto = new ExcelParseDto();
        excelParseDto.setName(csvFullName);
        excelParseDto.setGenDate(new SimpleDateFormat("yyyyMMdd").parse(genDate));
        excelParseDto.setStream(zip.getInputStream(zipEntry));
        DataMonitoringReportDto dataMonitoringReportDto = new DataMonitoringReportDto();
        try (InputStream inputStream =
            new FileInputStream(System.getProperty(SYS_URL) + File.separator + "application." + FILETYPE_PROPERTIES)) {
            Properties monitorConfig = new Properties();
            monitorConfig.load(inputStream);
            Map<String, List<DbMonitoringObjResultDto>> map = new HashMap<>();
            List<DbMonitoringObjResultDto> objResultDtoList = genDbConsistencyDtoList(excelParseDto.getStream());
            if ("consistencytablerows".equals(typeName)) {
                String uuid = monitorConfig.getProperty("monitor.target.datamodel.consistencytablerows.uuid." + idx);
                if (StringUtils.isBlank(uuid)) {
                    return;
                }
                String diffrence = monitorConfig.getProperty("monitor.target.consistency.rownum.difference." + idx);
                for (DbMonitoringObjResultDto dto : objResultDtoList) {
                    dto.setMonitoringObjectUuid(uuid);
                    int tRows =
                        Integer.valueOf(StringUtils.isEmpty(dto.getTargetTableRow()) ? "0" : dto.getTargetTableRow());
                    int sRows =
                        Integer.valueOf(StringUtils.isEmpty(dto.getSourceTableRow()) ? "0" : dto.getSourceTableRow());
                    int diff = tRows <= sRows ? (sRows - tRows) : (tRows - sRows);
                    boolean match =
                        (StringUtils.isNotEmpty(diffrence) && diff <= Integer.valueOf(diffrence)) ? true : false;
                    if (0 == diff) {
                        match = true;
                    }
                    dto.setMonitoringResult(match ? "pass" : "unpass");
                    String msg = "表记录相差条数:" + diff;
                    dto.setMonitoringResultDetail(match ? "" : msg);
                }
                map.put("1", objResultDtoList);
                dataMonitoringReportDto.setTableRowsMap(map);
                dataMonitoringReportDtos.add(dataMonitoringReportDto);
            } else if ("consistencytablefield".equals(typeName)) {
                String uuid = monitorConfig.getProperty("monitor.target.datamodel.consistencytablefield.uuid." + idx);
                if (StringUtils.isBlank(uuid)) {
                    return;
                }
                for (DbMonitoringObjResultDto dto : objResultDtoList) {
                    dto.setMonitoringObjectUuid(uuid);
                    String[] sFields = dto.getSourceFieldName().split("&");
                    List<String> sList = new ArrayList<>();
                    Collections.addAll(sList, sFields);
                    String[] tFields = dto.getTargetFieldName().split("&");
                    List<String> tList = new ArrayList<>();
                    Collections.addAll(tList, tFields);
                    List<String> res = new ArrayList<>();
                    for (String s : tList) {
                        if (!sList.contains(s)) {
                            res.add(s);
                        }
                    }
                    dto.setMonitoringResult(CollectionUtils.isEmpty(res) ? "pass" : "unpass");
                    String msg = "字段不一致:" + String.join(",", res);
                    dto.setMonitoringResultDetail(CollectionUtils.isEmpty(res) ? "" : msg);
                }
                map.put("1", objResultDtoList);
                dataMonitoringReportDto.setTableFieldMap(map);
                dataMonitoringReportDtos.add(dataMonitoringReportDto);
            }
        } catch (IOException e) {
            log.error(e.getMessage());
        }
    }

    private List<DbMonitoringObjResultDto> genDbConsistencyDtoList(InputStream stream) {
        List<DbMonitoringObjResultDto> resList = new ArrayList<>();
        EasyExcel.read(stream, DbConsistencyExcelDto.class, new ReadListener<DbConsistencyExcelDto>() {
            @Override
            public void invoke(DbConsistencyExcelDto data, AnalysisContext analysisContext) {
                DbMonitoringObjResultDto dbDto = new DbMonitoringObjResultDto();
                BeanUtils.copyProperties(data, dbDto);
                resList.add(dbDto);
            }

            // 表头校验
            @Override
            public void invokeHead(Map<Integer, ReadCellData<?>> headMap, AnalysisContext context) {
            }

            @Override
            public void doAfterAllAnalysed(AnalysisContext analysisContext) {
            }
        }).excelType(ExcelTypeEnum.CSV).autoTrim(true).headRowNumber(1).sheet().doRead();
        return resList;
    }

    @Override
    @SneakyThrows
    public void reportMonitorConsistencyData() {
        log.info("report monitor consistency data now");
        InputStream in = null;
        try {
            Properties dataSourceConfig = new Properties();
            File file = new File(System.getProperty(SYS_URL) + File.separator + "application." + FILETYPE_PROPERTIES);
            in = new FileInputStream(file);
            dataSourceConfig.load(in);
            DataParseDto dto = new DataParseDto();
            dto.setAccessCode(dataSourceConfig.getProperty("monitor.access.code"));
            dto.setProjectId(dataSourceConfig.getProperty("monitor.project.id"));
            dto.setUrl(dataSourceConfig.getProperty("monitor.haydn.url"));
            consistencyParseAndUpload(dto);
            log.info("report monitor consistency data succeed!");
        } catch (IOException e) {
            log.error(e.getMessage());
        } finally {
            if (in != null) {
                in.close();
            }
        }
    }
}
